package com.sh.diy;

import java.security.spec.RSAOtherPrimeInfo;
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @Auther: sh
 * @Version 1.0
 * @Description: 有界阻塞队列
 */
public class MyBlockQueue<T> {
    // 创建一个队列：Deque有很多实现类，大多数情况下使用ArrayDeque
    private Deque<T> deque = new ArrayDeque<>();
    // 队列的容量
    private int size;



    /*
        为什么定义锁？
        生产者往阻塞队列去添加任务，消费者从队头拿任务
        线程（生产者/消费者）可能有很多，任务（任务队列）只有一个，多线程去共享我们的一个资源，就会带来线程安全的问题
        */
    private ReentrantLock lock = new ReentrantLock();

    /*
   有了锁之后，定义两个条件变量，干嘛？
   当消费者去队列拿任务的时候，如果队列是空的，就让这个线程去休息室等待
   当生产者往队列添加任务的时候，如果队列是满的，就让这个线程去休息室等待
    */
    // 空的休息室  消费者
    Condition emptyCondition = lock.newCondition();
    // 满的休息室  生产者
    Condition fullCondition = lock.newCondition();
    // 添加任务
    // 阻塞添加，假如队列一直是满的，就一直等待
    public MyBlockQueue(int size) {
        this.size = size;
    }
    /* 定义方法：从队列中拿任务、给队列里面放任务 */
    public void put(T task) {
        //添加的时候，生产者可能有多个，而队列只有一个，那多线程访问共享资源，可能会有线程安全的问题，所以需要加锁
        lock.lock();

        try {
            // 如果有界队列满了，就到休息室去等待
            while (deque.size() == size) {
                System.out.println("===================队列满了，生产者等待================");
                fullCondition.await();
            }
            deque.addLast(task);
            System.out.println("生产者添加任务：" + task);
            // 唤醒消费者
            emptyCondition.signal();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            lock.unlock();
        }
    }

    //上面是阻塞添加，这个是带超时时间阻塞添加
    public boolean offer(T task, long timeout, TimeUnit timeUnit) {
        lock.lock();

        //把时间转为纳秒：以便后续使用awaitNanos方法
        long nanos = timeUnit.toNanos(timeout);
        // 有界队列满了
        try {
            while (deque.size() == size) {
                try {
                    if (nanos <= 0) {
                        System.out.println("====== 队列已满，生产者已等待" + timeUnit.toSeconds(timeout) + "秒，已超时...... ======");
                        return false;
                    }
                    System.out.println("task 等待加入任务队列...");
                    nanos = fullCondition.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            // 加到队列最后
            deque.addLast(task);
            System.out.println("task 加入任务队列成功 " + task);
            emptyCondition.signal();
            return true;
        } finally {
            lock.unlock();
        }
    }

    // 获取任务
    // 阻塞获取，假如队列一直是空的，就一直等待
    public T take() {
        lock.lock();
        try {
            while (deque.isEmpty()) {
                System.out.println("========队列是空的，消费者等待========");
                try {
                    emptyCondition.await();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            // 拿出第一个
            T t = deque.removeFirst();
            // 唤醒生产者
            fullCondition.signal();
            System.out.println("消费者拿到任务：" + t);
            return t;
        } finally {
            lock.unlock();
        }
    }

    //上面是阻塞获取，这个是带超时时间阻塞获取
    public T poll(long timeout, TimeUnit timeUnit) {
        lock.lock();
        try {
            //把时间转为纳秒：以便后续使用awaitNanos方法
            long nanos = timeUnit.toNanos(timeout);
            //队列为空
            while (deque.isEmpty()) {
                try {
                    if (nanos <= 0) {
                        System.out.println("====== 队列为空，消费者已等待" + timeUnit.toSeconds(timeout) + "秒，已超时...... ======");
                        return null;
                    }
                    System.out.println("等待获取任务 task...");
                    nanos = emptyCondition.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T t = deque.removeFirst();
            //唤醒生产者线程
            fullCondition.signal();
            System.out.println("获取了任务: " + t);
            return t;
        } finally {
            lock.unlock();
        }
    }

    public void tryPut(MyRejectPolicy<T> myRejectPolicy, T task) {
        lock.lock();
        try {
            if (deque.size() == size) {
                myRejectPolicy.reject(this, task);
            } else {
                deque.addLast(task);
                System.out.println("生产者添加任务：" + task);
                emptyCondition.signal();
            }
        } finally {
            lock.unlock();
        }
    }

}

